package com.boot.cluster;

import com.boot.cluster.listener.DispatchMessageListener;
import com.boot.cluster.listener.MemberStateListener;
import com.boot.cluster.listener.StateMachineHandle;
import org.apache.commons.lang3.StringUtils;
import org.jgroups.Address;
import org.jgroups.JChannel;
import org.jgroups.blocks.RpcDispatcher;
import org.jgroups.conf.ProtocolConfiguration;
import org.jgroups.conf.XmlConfigurator;
import org.jgroups.protocols.FD_SOCK;
import org.jgroups.protocols.TCP;
import org.jgroups.protocols.TCPPING;
import org.jgroups.protocols.VERIFY_SUSPECT;
import org.jgroups.raft.RaftHandle;
import org.jgroups.stack.IpAddress;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.ResourceUtils;

import java.io.File;
import java.io.FileInputStream;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;

public class ClusterRaftClient extends AbstractClusterClient {

    private static Logger logger = LoggerFactory.getLogger(ClusterRaftClient.class);

    private MemberStateListener stateListener;

    public ClusterRaftClient(ClusterConfig config) {
        this.config = config;
    }


    private List<IpAddress> cluster(ClusterConfig config) {
        if (StringUtils.isBlank(config.getClusterMembers())) {
            throw new IllegalArgumentException("Cluster member list is not set");
        }
        List<IpAddress> list = new ArrayList<>();
        Arrays.asList(config.getClusterMembers().split("\\,")).forEach((k) -> {
            for (int i = 0; i < ClusterConfig.scope; i++) {
                try {
                    list.add(new IpAddress(k, config.getPort() + i));
                } catch (Exception e) {
                    logger.warn("", e);
                }
            }
        });
        return list;
    }

    @SuppressWarnings("resource")
    @Override
    public void init() {
        try {
            XmlConfigurator configurator = XmlConfigurator.getInstance(new FileInputStream(new File(ResourceUtils.getURL("classpath:" + this.config.getConfig()).toURI())));
            ProtocolConfiguration config = configurator.getProtocolStack().stream().filter(s -> s.getProtocolName().equals("raft.RAFT")).findFirst().get();
            config.getProperties().put("raft_id", this.config.getMember());
            config.getProperties().put("members", this.config.getMember());

            this.channel = new JChannel(configurator).name(this.config.getMember());
            this.channel.getProtocolStack().findProtocol(TCP.class).setValue(
                    BIND_HOST_PARAM_NAME, InetAddress.getByName(this.config.getBindAddr()));

            //重置VERIFY_SUSPECT bind_addr属性
            this.channel.getProtocolStack().findProtocol(VERIFY_SUSPECT.class)
                    .setValue(BIND_HOST_PARAM_NAME, InetAddress.getByName(this.config.getBindAddr()));

            //重置FD_SOCK bind_addr属性
            this.channel.getProtocolStack().findProtocol(FD_SOCK.class)
                    .setValue(BIND_HOST_PARAM_NAME, InetAddress.getByName(this.config.getBindAddr()));

            //设置成员要发现的成员列表，集群成员的ip 逗号分隔
            this.channel.getProtocolStack().findProtocol(TCPPING.class)
                    .setValue("initial_hosts", this.cluster(this.config));

            //重置RAFT  members，支持动态添加(rp.addServer();rp.removeServer()) ，自动发现新节点
//			this.channel.getProtocolStack().findProtocol(RAFT.class)
//					.setValue("members",
//							//Arrays.asList(this.config.getRaftMembers().split("\\,")));
//							Arrays.asList(this.config.getMember())).setValue("raft_id", this.channel.name());

            this.raft = new RaftHandle(this.channel, new StateMachineHandle());
            this.channel.printProtocolSpec(true);
            this.raft.raftId(this.channel.name())
                    .addRoleListener(role -> {
                        logger.info("Current Node [{}],role:[{}]", this.raft.raftId(), role);
                        if (this.stateListener != null) {
                            this.stateListener.stateChanged(role);
                        }
                    });
            this.channel.setReceiver(new DispatchMessageListener(this));
            //是否接收本机发出的消息
            this.channel.setDiscardOwnMessages(false);
            this.channel.connect(this.config.getClustName());
        } catch (Exception e) {
            logger.error("cluster init error ", e);
        }
    }


    public RpcDispatcher getRpcDispatcher() {
        return new RpcDispatcher(this.channel, null);
    }

    public List<Address> getMember() {
        return this.channel.getView().getMembers();
    }

    public void setStateListener(MemberStateListener handle) {
        this.stateListener = handle;
    }

    /***
     * 当前节点的角色
     * @return
     */
    public String getRole() {
        return this.raft.raft().role();
    }
}
